package com.bw.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.bean.PageTrafficStats;
import com.bw.gmall.realtime.utils.DateFormatUtil;
import com.bw.gmall.realtime.utils.MyClickHouseUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 页面流量TOP10统计应用
 * 统计访客数最高的前10个页面的数据情况
 * - PageTrafficTopNApp.java ：页面访问TOP N统计
 * -
 */
/**
 * 流量域页面访问TopN应用
 * 负责统计网站各页面的访问量TopN数据，分析页面热度分布
 * 主要功能点：
 * - 从Kafka的ods_traffic主题读取页面日志数据
 * - 使用Flink水印策略和窗口处理
 * - 统计各个页面的访问量
 * - 计算TopN热门页面
 * - 使用ClickHouse工具类进行数据存储或查询
 * 
 * 数据流向：
 * 页面日志数据 -> PageTrafficTopNApp -> 处理后输出热门页面TopN结果
 */
public class PageTrafficTopNApp {
    public static void main(String[] args) throws Exception {
        // 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 2. 读取Kafka的页面日志主题
        String pageTopic = "ods_traffic";
        String groupId = "page_traffic_topn_group";
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getFlinkKafkaConsumer(pageTopic, groupId);
        SingleOutputStreamOperator<JSONObject> jsonObjStream = env.addSource(kafkaSource)
                .map(new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String s) throws Exception {
                        return JSON.parseObject(s);
                    }
                }).assignTimestampsAndWatermarks(
                        WatermarkStrategy.<JSONObject>
                                forBoundedOutOfOrderness(Duration.ofSeconds(10))
                                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject jsonObject, long l) {
                                        return jsonObject.getLong("ts");
                                    }
                                })
                );

        // 3. 提取页面访问信息
        SingleOutputStreamOperator<PageTrafficStats> pageStatsStream = jsonObjStream.map(new MapFunction<JSONObject, PageTrafficStats>() {
            @Override
            public PageTrafficStats map(JSONObject jsonObject) throws Exception {
                String statsTime = DateFormatUtil.toYmdHms(jsonObject.getLong("ts"));                String statsType = "10s";
                
                // 获取页面信息
                JSONObject page = jsonObject.getJSONObject("page");
                String pageId = page.getString("page_id");
                String pageName = getPageName(pageId); // 根据page_id获取页面名称
                long stayTime = page.containsKey("stay_time") ? page.getLong("stay_time") : 0L;
                
                // 初始化统计指标
                PageTrafficStats stats = new PageTrafficStats();
                stats.setStatsTime(statsTime);
                stats.setStatsType(statsType);
                stats.setPageId(pageId);
                stats.setPageName(pageName);
                stats.setUvCount(1L); // 每个mid算作一个访客
                stats.setPvCount(1L); // 每个页面访问算作一个PV
                stats.setStayTime(stayTime);
                
                // 计算跳失率（简单计算：只有一个页面访问的会话视为跳失）
                // 这里简化处理，实际项目中需要更复杂的会话分析
                double bounceRate = 0.0;
                if (stayTime < 5 && !jsonObject.containsKey("actions")) {
                    bounceRate = 1.0;
                }
                
                // 引导到商品页次数
                long guideToGoodsCount = 0L;
                if (jsonObject.containsKey("actions")) {
                    JSONArray actions = jsonObject.getJSONArray("actions");
                    for (Object action : actions) {
                        JSONObject actionObj = (JSONObject) action;
                        String actionType = actionObj.getString("action").toLowerCase();
                        if ("click".equals(actionType)) {
                            String itemType = actionObj.containsKey("item_type") ? actionObj.getString("item_type") : "";
                            if ("sku_id".equals(itemType)) {
                                guideToGoodsCount = 1L;
                                break;
                            }
                        }
                    }
                }
                // 统计点击分布（简单示例）
                String clickDistribution = getClickDistribution(jsonObject);
                
                stats.setBounceRate(bounceRate);
                stats.setGuideToGoodsCount(guideToGoodsCount);
                stats.setClickDistribution(clickDistribution);
                stats.setCreateTime(new Date());
                
                return stats;
            }
        });

        // 4. 按页面ID和统计时间分组，统计指标
        SingleOutputStreamOperator<PageTrafficStats> aggregatedStream = pageStatsStream
                .keyBy(new KeySelector<PageTrafficStats, Tuple3<String, String, String>>() {
                    @Override
                    public Tuple3<String, String, String> getKey(PageTrafficStats stats) throws Exception {
                        return new Tuple3<>(stats.getStatsTime(), stats.getStatsType(), stats.getPageId());
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.minutes(5)))
                .reduce(new ReduceFunction<PageTrafficStats>() {
                    @Override
                    public PageTrafficStats reduce(PageTrafficStats value1, PageTrafficStats value2) throws Exception {
                        // 累加各项指标
                        value1.setUvCount(value1.getUvCount() + value2.getUvCount());
                        value1.setPvCount(value1.getPvCount() + value2.getPvCount());
                        value1.setStayTime(value1.getStayTime() + value2.getStayTime());
                        // 跳失率计算：总跳失数/总访问数
                        // 注意：这里是简化计算，实际项目中应分别统计跳失数和总访问数
                        double totalBounceRate = (value1.getBounceRate() + value2.getBounceRate()) / 2;
                        value1.setBounceRate(totalBounceRate);
                        value1.setGuideToGoodsCount(value1.getGuideToGoodsCount() + value2.getGuideToGoodsCount());
                        // 点击分布合并（简化处理）
                        value1.setClickDistribution(mergeClickDistributions(value1.getClickDistribution(), 
                                                                          value2.getClickDistribution()));
                        return value1;
                    }
                });

        // 5. 按统计时间和统计类型分组，计算页面TOP10
        SingleOutputStreamOperator<PageTrafficStats> top10Stream = aggregatedStream
                .keyBy(new KeySelector<PageTrafficStats, Tuple3<String, String, String>>() {
                    @Override
                    public Tuple3<String, String, String> getKey(PageTrafficStats stats) throws Exception {
                        return new Tuple3<>(stats.getStatsTime(), stats.getStatsType(), "page_topn");
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.minutes(5)))
                .apply(new WindowFunction<PageTrafficStats, PageTrafficStats, Tuple3<String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple3<String, String, String> key, TimeWindow window,
                                     Iterable<PageTrafficStats> input, Collector<PageTrafficStats> out) throws Exception {
                        // 将窗口内的数据转换为列表
                        List<PageTrafficStats> statsList = new ArrayList<>();
                        for (PageTrafficStats stats : input) {
                            statsList.add(stats);
                        }

                        // 按访客数降序排序，取TOP10
                        List<PageTrafficStats> top10List = statsList.stream()
                                .sorted(Comparator.comparing(PageTrafficStats::getUvCount).reversed())
                                .limit(10)
                                .collect(Collectors.toList());

                        // 为TOP10数据添加排名
                        for (int i = 0; i < top10List.size(); i++) {
                            PageTrafficStats stats = top10List.get(i);
                            stats.setRank(i + 1);
                            out.collect(stats);
                        }
                    }
                });
        // 6. 将结果写入ClickHouse
        top10Stream.addSink(MyClickHouseUtil.getSinkFunction(
                "insert into dws_page_traffic_topn_window values(?,?,?,?,?,?,?,?,?,?,?,?)"
        ));
        // 7. 执行作业
        env.execute("PageTrafficTopNApp");
    }

    /**
     * 根据page_id获取页面名称
     */
    private static String getPageName(String pageId) {
        Map<String, String> pageNameMap = new HashMap<>();
        pageNameMap.put("home", "首页");
        pageNameMap.put("category", "分类页");
        pageNameMap.put("search", "搜索页");
        pageNameMap.put("good_detail", "商品详情页");
        pageNameMap.put("cart", "购物车");
        pageNameMap.put("order_confirm", "订单确认页");
        pageNameMap.put("payment", "支付页");
        pageNameMap.put("payment_success", "支付成功页");
        pageNameMap.put("my_orders", "我的订单页");
        pageNameMap.put("user_center", "用户中心");
        
        return pageNameMap.getOrDefault(pageId, "未知页面");
    }

    /**
     * 获取页面点击分布数据
     */
    private static String getClickDistribution(JSONObject jsonObject) {
        // 简化处理，实际项目中应根据页面的具体区块统计点击分布
        Map<String, Long> clickMap = new HashMap<>();
        
        if (jsonObject.containsKey("actions")) {
            JSONArray actions = jsonObject.getJSONArray("actions");
            for (Object action : actions) {
                JSONObject actionObj = (JSONObject) action;
                String actionType = actionObj.getString("action").toLowerCase();
                if ("click".equals(actionType)) {
                    String area = actionObj.containsKey("area") ? actionObj.getString("area") : "default";
                    clickMap.put(area, clickMap.getOrDefault(area, 0L) + 1);
                }
            }
        }
        
        // 如果没有点击数据，返回默认值
        if (clickMap.isEmpty()) {
            clickMap.put("default", 0L);
        }
        
        return JSON.toJSONString(clickMap);
    }

    /**
     * 合并点击分布数据
     */
    private static String mergeClickDistributions(String dist1, String dist2) {
        // 修复类型转换问题，确保Integer类型正确转换为Long类型
        Map<String, Object> map1 = JSON.parseObject(dist1, Map.class);
        Map<String, Object> map2 = JSON.parseObject(dist2, Map.class);
        
        Map<String, Long> mergedMap = new HashMap<>();
        
        // 处理第一个map
        for (Map.Entry<String, Object> entry : map1.entrySet()) {
            long count = parseLongValue(entry.getValue());
            mergedMap.put(entry.getKey(), count);
        }
        
        // 处理第二个map并合并
        for (Map.Entry<String, Object> entry : map2.entrySet()) {
            long count = parseLongValue(entry.getValue());
            mergedMap.put(entry.getKey(), mergedMap.getOrDefault(entry.getKey(), 0L) + count);
        }
        
        return JSON.toJSONString(mergedMap);
    }
    
    /**
     * 安全地将Object转换为Long
     */
    private static long parseLongValue(Object value) {
        if (value instanceof Long) {
            return (Long) value;
        } else if (value instanceof Integer) {
            return ((Integer) value).longValue();
        } else if (value instanceof String) {
            try {
                return Long.parseLong((String) value);
            } catch (NumberFormatException e) {
                return 0L;
            }
        }
        return 0L;
    }
}